package test.sunyu.storm.demo.demo1;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import sunyu.kit.common.LogKit;

/**
 * @author 孙宇
 */
public class TopologyMain {

    private static final Logger logger = LogKit.getLogger();

    public static void main(String[] args) {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("word-reader", new WordReader());
        builder.setBolt("word-normalizer", new WordNormalizer()).localOrShuffleGrouping("word-reader");
        builder.setBolt("word-counter", new WordCounter(), 2).fieldsGrouping("word-normalizer", new Fields("word"));

        Config conf = new Config();
        if (args.length == 1) {
            conf.put("wordsFile", args[0]);
        } else {
            String url = TopologyMain.class.getResource("/word.txt").getPath().substring(1);
            logger.debug("准备统计单词文件 {}", url);
            conf.put("wordsFile", url);
        }
        //conf.setDebug(false);
        conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);

        logger.info("正在提交拓扑到本地处理");

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("Getting-Started-Topologie", conf, builder.createTopology());
        Utils.sleep(1000 * 20);
        cluster.killTopology("Getting-Started-Topologie");
        cluster.shutdown();
    }

}
